/*
 * Package il.ac.biu.cs.grossmm.impl.activeData
 * File SubscriptionDispatcher.java
 * 
 * This is a part of presence service framework prototype. 
 * See javadoc for more information.
 *  
 * Copyright (C) 2006 Grossmann Mark
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.

 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */
package il.ac.biu.cs.grossmm.impl.activeData;

import static il.ac.biu.cs.grossmm.impl.keys.PatternMatcher.*;
import static il.ac.biu.cs.grossmm.impl.keys.PatternOperations.*;
import il.ac.biu.cs.grossmm.api.flow.NotificationProvider;
import il.ac.biu.cs.grossmm.api.flow.Status;
import il.ac.biu.cs.grossmm.api.flow.Track;
import il.ac.biu.cs.grossmm.api.keys.Key;
import il.ac.biu.cs.grossmm.api.keys.KeyPattern;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Logger;

public class SubscriptionDispatcher<N> {
	static Logger logger = Logger.getLogger("SubscriptionDispatcher");

	/**
	 * Subscription point associated with this dispatcher
	 */
	SubscriptionPointImpl<N> subscriptionPoint;

	/**
	 * List of all links from this subscription point to all conditionally and
	 * unconditionally reachable output points (notification and publication
	 * points)
	 */
	List<DataPointLink<N>> links = new ArrayList<DataPointLink<N>>();

	/**
	 * Creates subscription dispatcher for given subscription point
	 * 
	 * @param subscriptionPoint
	 *            the subscription point to associated this object with
	 */
	public SubscriptionDispatcher(SubscriptionPointImpl<N> subscriptionPoint) {
		this.subscriptionPoint = subscriptionPoint;
	}

	/**
	 * Find the best possible subscription path an output point which can
	 * provide data for given key
	 * 
	 * @param key
	 *            the key to subscribe to
	 * @param track
	 *            subscription track
	 * @return
	 */
	Status subscribe(Key key, Track track) {
		SubscriptionImpl<N> subscription = null;

		// try all data-links in descending priority order

		for (DataPointLink<N> link : links) {

			// does the key match the pattern associated with this link?
			if (!link.matches(key))
				continue;

			// subscription = np.subscribe(key, track);
			subscription = link.subscribe(subscriptionPoint.subscriber, key,
					track);

			if (subscription != null)
				break;

			int removeme = 0;
			removeme--;
		}

		return subscription == null ? Status.UNMATCHED : subscription
				.getStatus();
	}

	/**
	 * Computes best conversion chains from the associated subscription point to
	 * all reachable output points from this subscription point
	 * 
	 */
	@SuppressWarnings("unchecked")
	public void updateLinks() {
		links.clear();

		// recurssion enterence point
		visit(dataLayer(), new HashSet(), subscriptionPoint.getPattern(), null);

		Collections.sort(links);
	}

	/**
	 * Given a conversion chain (possibly empty) from this subscription point to
	 * some conditionaly s-reachable layer (or this layer, if the chain is
	 * empty) find all conversion chains that extend the given chain and do not
	 * pass through any layer passed by the given chain (the visited layers)
	 * 
	 * @param xLayer
	 *            a conditionally reachable layer, or this layer
	 * @param visitedLayers
	 *            all layers the conversion chain passes
	 * @param pattern
	 *            the pattern which restricts the data keys, subscriptions for
	 *            which can use this chain
	 * @param baseChain
	 *            the abovementioned conversion chain
	 */
	@SuppressWarnings("unchecked")
	void visit(DataLayer xLayer, Set<DataLayer> visitedLayers,
			KeyPattern pattern, ConversionChain baseChain) {
		Map<DataLayer, ConversionChain> bestChains = xLayer.conversionChains;

		// set of all unconditionally s-reachable layer from the specified layer
		Set<DataLayer> ucRechableLayers = bestChains.keySet();

		// add the xLayer and all s-reachable layers from it to the set
		// of visited layers
		HashSet visitedLayers2 = new HashSet(visitedLayers);
		visitedLayers2.add(xLayer);
		visitedLayers2.addAll(ucRechableLayers);

		// extract links to all putput points on data layer
		// to which the associated subscription point belongs
		addOutputPoints(xLayer, baseChain, pattern);

		// process all conditionally reachable layers from xLayer
		// [this]~~~~>[xLayer]~~~~>(iterate)
		addConditionallyReachable(xLayer, baseChain, visitedLayers2, pattern);

		// for each unconditionally s-reachable layer
		for (DataLayer yLayer : ucRechableLayers) {

			// the best chain of converters to reach the u/c s-reachable layer
			ConversionChain bestChain = bestChains.get(yLayer);
			ConversionChain chain;

			if (baseChain == null)
				chain = bestChain;
			else
				chain = new ConversionChain(baseChain, bestChain);

			// add all reachable output points on this layer
			addOutputPoints(yLayer, chain, pattern);

			// process all conditionally reachable layers
			// [this]~~~~>[xLayer]--->[yLayer]~~~~>(iterate)
			addConditionallyReachable(yLayer, chain, visitedLayers2, pattern);
		}

	}

	/**
	 * Given an s-reachable layer (unconditionally or conditionally) add links
	 * to all output points on this layer
	 * 
	 * @param layer
	 *            the data layer
	 * @param chain
	 *            the conversion chain used to reach this layer
	 * @param pattern
	 *            the pattern which restricts the data keys, subscriptions for
	 *            which can use this chain
	 */
	private void addOutputPoints(DataLayer layer, ConversionChain chain,
			KeyPattern pattern1) {
		// get list of notification providers on the destination layer
		List<NotificationProvider> npList = layer.getNotificationProviders();

		// for all notification providers on the destination layer
		for (NotificationProvider np : npList) {

			KeyPattern pattern2 = np.getPattern();

			// the chain of unconditional converters for this link
			ConversionProcessor[] converters = chain == null ? null : chain
					.getConverters();

			// use pattern transformations if there are any conditional
			// converters in the chain
			if (chain != null && chain.isConditional()) {
				if (compare(pattern1, pattern2) == DISJOINT)
					continue;

				int i = converters.length;
				while (i-- > 0) {
					ConversionProcessor cp = converters[i];
					if (cp.isConditional()) {
						KeyPattern srcPattern = cp.getSourcePattern();
						KeyPattern dstPattern = cp.getDestinationPattern();
						pattern2 = transform(pattern2, srcPattern, dstPattern);
					}
				}
			}

			KeyPattern pattern = subscriptionPoint.getPattern();

			// check how input and output patterns relate to each other
			int relation = compare(pattern, pattern2);

			// if patterns are disjoint, do not add this link
			if (relation == DISJOINT)
				continue;

			if ((relation & LEFT_RESTRICTION) != LEFT_RESTRICTION)
				// use intersection of the patterns
				// to match keys
				pattern = intersection(pattern, pattern2);
			else
				// input pattern is a restriction of the
				// output pattern ==> any key will match
				pattern = null;

			// priority of this link
			double priority = (chain == null ? 1.0 : chain.getPriority())
					* np.getPriority();

			if (converters != null) {
				int t = 0;
				t++;
			}

			addLink(converters, np, pattern, priority);
		}
	}

	void addLink(ConversionProcessor[] converters, NotificationProvider np,
			KeyPattern pattern, double priority) {
		// create the link
		DataPointLink<N> link = new DataPointLink<N>(subscriptionPoint, np,
				converters, pattern, priority);

		links.add(link);

		// logger.info("Added " + link);
	}

	/**
	 * Given an unconditionally or conditionally s-reachable layer add links to
	 * all conditionally s-reachable output points on other layers, except the
	 * previously visited layers and unconditionally s-reachable layers from the
	 * previously visited layers
	 * 
	 * @param yLayer
	 *            the data layer
	 * @param chain
	 *            the conversion chain used to reach this layer
	 * @param visitedLayers
	 *            set of all previously visited layers (passed by the chain) and
	 *            all unconditionally s-reachable layers from the visited layers
	 * @param pattern
	 *            the pattern which restricts the data keys, subscriptions for
	 *            which can use this chain
	 */
	private void addConditionallyReachable(DataLayer yLayer,
			ConversionChain chain, Set<DataLayer> visitedLayers,
			KeyPattern pattern1) {
		// set of all layers for which there are conditional conveters
		// to the yLayer (i.e. set of all directly conditionally s-reachable
		// layers_
		Set<DataLayer> srcLayers = yLayer.getConditionallyReachableLayers();

		// for each such layer
		for (DataLayer srcLayer : srcLayers) {
			// except already visited / conditionally s-reachable from visited
			if (visitedLayers.contains(srcLayer))
				continue;

			// set of conditional converters to s-reach this layer
			Set<ConversionProcessor> convertors = yLayer
					.getConditionals(srcLayer);

			// for such converters
			for (ConversionProcessor convertor : convertors) {
				assert convertor.isConditional();

				KeyPattern srcPattern = convertor.getSourcePattern();
				KeyPattern dstPattern = convertor.getDestinationPattern();

				// do input pattern and the converter source pattern intersect?
				if (compare(pattern1, dstPattern) == DISJOINT)
					continue; // no => cannot use the converter in this chain

				// transform the pattern according to the converter's conditions
				KeyPattern pattern2 = transform(pattern1, dstPattern,
						srcPattern);

				// append the chain to reach the dstLayer
				ConversionChain chain2 = new ConversionChain(chain, convertor);

				// recursively add all links
				visit(srcLayer, visitedLayers, pattern2, chain2);
			}
		}
	}

	/**
	 * @return
	 */
	private DataLayer<N> dataLayer() {
		return subscriptionPoint.dataLayer;
	}
}
